package com.gitee.swsk33.kafkastreamreceiver.config;

import com.gitee.swsk33.kafkaentity.dataobject.Precipitation;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;

import java.time.LocalDateTime;

/**
 * Kafka流处理配置
 */
@Configuration
@EnableKafkaStreams
public class KafkaStreamConfig {

	@Bean
	public KStream<String, Precipitation> kStream(StreamsBuilder streamsBuilder) {
		// 表示接受weather话题的消息并构建为流
		KStream<String, Precipitation> stream = streamsBuilder.stream("weather");
		// 执行流操作
		stream.map((key, value) -> {
			// 流映射：将降水量数据中的值由单位0.1mm转换成1mm
			value.setValue(value.getValue() / 10);
			return new KeyValue<>(key, value);
		}).filter((key, value) -> {
			// 流过滤：去除小于2mm的数据
			// 设定处理时间
			value.setGmtModified(LocalDateTime.now());
			return value.getValue() > 2;
		}).foreach((key, value) -> {
			System.out.println(value);
		});
		return stream;
	}

}